package com.zyx.pulsardemo.connector.sink;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;

import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;

/**
 * @author Yaxi.Zhang
 * @since 2021/10/12 14:32
 * desc: Sink连接器, 接受数据写入文件中
 *       该Sink连接器有一些限制, 特别是不能将无穷无尽的数据流写入单个文件
 */
public class LocalFileSink implements Sink<String> {
    public static final String FILENAME_PREFIX_CONFIG_KEY = "filenamePrefix";
    public static final String FILENAME_SUFFIX_CONFIG_KEY = "filenameSuffix";
    private BufferedWriter bw = null;
    private FileWriter fw = null;

    @Override
    public void close() throws Exception {
        try {
            if (bw != null) {
                bw.close();
            }
            if (fw != null) {
                fw.close();
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    @Override
    public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
        // 从提供的配置 属性检索目标文件名前缀和后缀
        String filenamePrefix = (String) config.getOrDefault(FILENAME_PREFIX_CONFIG_KEY, "test-out");
        String filenameSuffix = (String) config.getOrDefault(FILENAME_SUFFIX_CONFIG_KEY, ".tmp");
        // 在临时目录中创建新文件
        File file = File.createTempFile(filenamePrefix, filenameSuffix);
        // 初始化文件和缓冲写入器
        fw = new FileWriter(file.getAbsoluteFile(), true);
        bw = new BufferedWriter(fw);
    }

    @Override
    public void write(Record<String> record) throws Exception {
        try {
            // 从传入的记录中检索值并将其写入打开的文件
            bw.write(record.getValue());
            bw.flush();
            // 确认我们成功地处理了消息, 以便可以清除它
            record.ack();
        } catch (IOException e) {
            // 表示我们无法处理该消息, 以便可以保留该消息并在稍后重试
            record.fail();
            throw new RuntimeException(e);
        }
    }
}
